package com.example.demo.stream.producer;

import com.example.demo.stream.common.DataObject;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class StreamMessageProducer {

    private final StringRedisTemplate stringRedisTemplate;

    public StreamMessageProducer(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public void send(String stream, DataObject<?> obj) {
        ObjectMapper mapper = new ObjectMapper();

        // DataObject → Map<String, String>
        Map<String, String> body = mapper.convertValue(obj, new TypeReference<Map<String, String>>() {});

        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(stream)
                .ofMap(body);

        stringRedisTemplate.opsForStream().add(record);
    }
}
